package com.distributed.netty.server;

import com.distributed.netty.entity.*;
import com.rcache.core.utils.EmailUtils;
import com.rcache.core.cache.Rcache;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.apache.log4j.Logger;

import javax.mail.MessagingException;
import java.io.UnsupportedEncodingException;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Objects;

/**
 * @desc 消息处理手柄
 * @author gzy
 * @time 2022-06-05
 */
public class ChatHandler extends SimpleChannelInboundHandler<Object> {
    private Rcache rcache;
    private LinkedList<ElementJob> jobList;
    public ChatHandler(Rcache rcache, LinkedList<ElementJob> jobList) {
        this.rcache=rcache;
        this.jobList=jobList;
    }
    Logger LOG=Logger.getLogger(ChatHandler.class);
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        System.err.println(Cache.set);
        System.err.println(Cache.cache);
        Message message = (Message) msg;
        HashSet<ElementJob> jobs = message.getElementJobs();
        if (jobs!=null){
            for (ElementJob job:jobs) {
                switch (job.getJobType()){
                    case PUT:
                        rcache.put(job.getArgs()[0],job.getArgs()[1]);
                        LOG.debug("master执行slave的 put 操作……");
                        break;
                    case DELETE:
                        rcache.remove(job.getArgs()[0]);
                        LOG.debug("master执行slave的 remove 操作……");
                        break;
                    case UPDATE:
                        rcache.replace(job.getArgs()[0],job.getArgs()[1]);
                        LOG.debug("master执行slave的 replace 操作……");
                        break;
                }
            }
            //回复消息
            Message replyMessage=new Message();
            LOG.debug("接收到的消息："+message.getMsg());
            //对发送来的slave记录其ip 发送job在slave端验证如果是自己则删除job
            String ip = channelHandlerContext.channel().remoteAddress().toString().split(":")[0].replace("/","");
            replyMessage.setSendSlaveIp(ip);
            LOG.debug("本次请求的slaveIp为："+ip);
            replyMessage.setSlaves(Cache.set);
            replyMessage.setMsg("更新操作");
            replyMessage.setElementJobs(jobs);
            replyMessage.setSendSlaveIp(ip);
            LOG.debug("master 轮训向slave发送更新操作");
            for (String slaveIp:Cache.cache.keySet()) {
                Cache.cache.get(slaveIp).getChannelHandlerContext().writeAndFlush(replyMessage);
            }
            jobs.clear();
        }else {
            if (!jobList.isEmpty()){
                //回复消息
                Message replyMessage=new Message();
                LOG.debug("接收到的消息："+message.getMsg());
                LOG.debug("将服务端的job发送给所有的slave 任务列表：");
                LOG.debug(jobList);
                replyMessage.setSlaves(Cache.set);
                replyMessage.setMsg("更新操作");
                HashSet jobs2=new HashSet();
                jobs2.addAll(jobList);
                replyMessage.setElementJobs(jobs2);
                for (String slaveIp:Cache.cache.keySet()) {
                    Cache.cache.get(slaveIp).getChannelHandlerContext().writeAndFlush(replyMessage);
                }
                jobList.clear();
            }else {
                //回复心跳，将slaves 发送
                Message replyMessage=new Message();
                LOG.debug("接收到的心跳消息："+message.getMsg());
                LOG.debug("将服务端的slaves发送给所有的slave");
                replyMessage.setSlaves(Cache.set);
                replyMessage.setMsg("更新slaves");
                for (String slaveIp:Cache.cache.keySet()) {
                    Cache.cache.get(slaveIp).getChannelHandlerContext().writeAndFlush(replyMessage);
                }
            }

        }

        Slave slave=new Slave(channelHandlerContext);
        String slaveIp = channelHandlerContext.channel().remoteAddress().toString();
        Cache.cache.put(slaveIp,slave);
        System.out.println(Cache.set);
        boolean flag = Cache.set.add(slaveIp);
        System.out.println(flag);
        System.out.println(Cache.set);
        if (flag){
            LOG.debug("新连接一个slave，将本机的缓存制作成job发送给这个slave");
            Message replyMessage=new Message();
            HashSet set=new HashSet();
            for (Object key:rcache.KeySet()) {
                Object value = rcache.get(key);
                ElementJob job = new ElementJob(JobType.PUT, new Object[]{key, value}, "接收服务端的缓存");
                set.add(job);
            }
            replyMessage.setMsg("接收服务端的缓存");
            replyMessage.setElementJobs(set);

            channelHandlerContext.writeAndFlush(replyMessage);
        }
    }
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        LOG.debug("客户端断开, channel对应的长id为:" + ctx.channel().id().asLongText());
        LOG.debug("客户端断开, channel对应的短id为:" + ctx.channel().id().asShortText());
        String ip = ctx.channel().remoteAddress().toString();
        Cache.set.remove(ip);
        Cache.cache.remove(ip);
        LOG.debug("将slaves中移除改终端:"+ip);
        //如果开启了预警系统则异步发送预警信息
        if (rcache.isWarning()){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    //获取本机ip
                    InetAddress inet4Address = null;
                    try {
                        inet4Address = Inet4Address.getLocalHost();
                        String ipSelf = inet4Address.toString().split("/")[1];
                        //发送告警信息
                        EmailUtils.createMineMessage("master:"+ipSelf+"的预警信息","ip地址为:"+ip.split(":")[0].replace("/","")+"的slave终端宕机断线",rcache.getManagerEmail());
                    } catch (UnknownHostException | UnsupportedEncodingException | MessagingException e) {
                        throw new RuntimeException(e);
                    }
                }
            }).start();
        }
    }
}